package sparkStreaming

import java.io.{BufferedReader, InputStreamReader}
import java.net.Socket
import java.nio.charset.StandardCharsets

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.receiver.Receiver



object Driver{
	def main(args: Array[String]): Unit = {
		val conf: SparkConf = new SparkConf().setAppName("wordcount").setMaster("local[*]")

		// 构建 StreamingContext
		val streamingContext = new StreamingContext(conf, Seconds(3))

		// 从指定的端口采集数据,
		val socketLineDStream: ReceiverInputDStream[String] = streamingContext.receiverStream(new CustomerReceiver("localhost", 9999))

		// 将采集的数据进行分解
		val wordDStream: DStream[String] = socketLineDStream.flatMap(_.split(" "))

		// 将结构转换便于统计
		val mapDStream: DStream[(String, Int)] = wordDStream.map((_, 1))

		// reduceByKey 聚合
		val sumDStream: DStream[(String, Int)] = mapDStream.reduceByKey(_+_)

		// 打印结果，在Driver打印
		sumDStream.print()

		// 不能停止采集程序
		//streamingContext.stop()

		// 启动接收器
		streamingContext.start()
		// Driver等待接收器停止
		streamingContext.awaitTermination()
	}
}



class CustomerReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_ONLY) {

	//最初启动的时候，调用该方法，作用为：读数据并将数据发送给Spark
	override def onStart(): Unit = {
		new Thread("Socket Receiver") {
			override def run() {
				receive()
			}
		}.start()
	}

	//读数据并将数据发送给Spark
	def receive(): Unit = {

		//创建一个Socket
		var socket: Socket = new Socket(host, port)

		//定义一个变量，用来接收端口传过来的数据
		var input: String = null

		//创建一个BufferedReader用于读取端口传来的数据
		val reader = new BufferedReader(new InputStreamReader(socket.getInputStream, StandardCharsets.UTF_8))

		//读取数据
		input = reader.readLine()

		//当receiver没有关闭并且输入数据不为空，则循环发送数据给Spark
		while (!isStopped() && input != null) {
			store(input)
			input = reader.readLine()
		}

		//跳出循环则关闭资源
		reader.close()
		socket.close()

		//重启任务
		restart("restart")
	}

	override def onStop(): Unit = {
		println("stop receive")
	}
}
